package com

import org.apache.hadoop.io.{IntWritable, Text}
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object WordCount1 {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root")
    // 1. 创建StreamingContext
    val conf = new SparkConf().setMaster("local[2]").setAppName("WordCount1")
    val ssc = new StreamingContext(conf, Seconds(3)) //设置每次处理批次,比如说三秒处理一次.就设置为3
    // 2. 从数据源创建一个流:  监听指定端口的数据
    val sourceStream = ssc.socketTextStream("zjj101", 9999)
    // 3. 对流做各种转换
    // wordcount程序
    val result = sourceStream.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_ + _)
    result.print() //打印到控制台上.
    //        val result = sourceStream.flatMap(x => {Thread.sleep(10000); x.split(" ")}).map((_, 1)).reduceByKey(_ + _)
    //     4. 行动算子 print  foreach foreachRDD
    //            result.print() // 把结果打印在控制台
    //        result.saveAsNewAPIHadoopFiles("/1015/test",
    //          "wc",
    //          classOf[Text],
    //          classOf[IntWritable],
    //          classOf[TextOutputFormat[Text, IntWritable]])
    // 5. 启动流
    ssc.start()
    // 6. 阻止主线程退出(阻塞主线程)
    ssc.awaitTermination()
  }
}
